package com.imooc.bigdata.chapter08_my.business

import com.imooc.bigdata.chapter08.`trait`.DataProcess
import com.imooc.bigdata.chapter08.utils.KuduUtils
import com.imooc.bigdata.chapter08_my.utils.{DateUtils, SQLUtils, SchemaUtils}
import org.apache.kudu.Schema
import org.apache.spark.sql.{DataFrame, SparkSession}

object ProvinceCityStatProcess extends DataProcess{
  override def process(spark: SparkSession): Unit = {
    val SOURCE_TABLE_NAME = DateUtils.getTableName("ods",spark)
    val MASTER="spark000"
    //从kudu 的ODS表中读取数据，然后按照省份和城市分组统计
    var odsDF: DataFrame = spark.read.format("org.apache.kudu.spark.kudu")
      .option("kudu.table", SOURCE_TABLE_NAME)
      .option("kudu.master", MASTER)
      .load()

    // 扩展：求每个省份下每个城市数量最多的TopN   这是一个典型的TopN


    //写sql得到结果
    odsDF.createOrReplaceTempView("ods")
    var resultFD: DataFrame = spark.sql(SQLUtils.PROVINCE_CITY_SQL)



    //写回kudu
    val SINK_TABLE_NAME =DateUtils.getTableName("province_city_stat",spark)
    val partitionId = "provincename"
    var schema: Schema = SchemaUtils.ProvinceCitySchema

    KuduUtils.sink(resultFD,SINK_TABLE_NAME,MASTER,schema,partitionId)
  }
}
